package io.kiki.sba.registry.store.api.elector;

import io.kiki.sba.registry.net.NetUtil;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.LoopExecuteTask;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;

@Setter
@Getter
public abstract class AbstractLeaderElector implements LeaderElector {
    private static final Logger LOG = LoggerFactory.getLogger(AbstractLeaderElector.class);

    private final List<LeaderAware> leaderAwareList = new CopyOnWriteArrayList<>();
    private final LeaderElectorTrigger leaderElectorTrigger = new LeaderElectorTrigger();
    private volatile Leader leader = Leader.HAS_NO_LEADER;
    private volatile boolean startElector = false;
    private boolean isObserver = false;
    @Setter
    private volatile String address;

    public AbstractLeaderElector() {
        address = NetUtil.getLocalAddress().getHostAddress();
    }

    protected static Leader calcLeaderInfo(String leader, long epoch, long lastHeartbeat, long duration) {
        final long expireTimestamp = lastHeartbeat + duration / 2;
        return new Leader(leader, epoch, expireTimestamp);
    }

    @Override
    public void registerLeaderAware(LeaderAware leaderAware) {
        leaderAwareList.add(leaderAware);
    }

    @PostConstruct
    public void init() {
        ConcurrentUtils.createDaemonThread("LeaderElectorTrigger", leaderElectorTrigger).start();
    }


    public void elect() {
        synchronized (this) {
            if (isObserver) { // 如果是Observer，不参与选主
                leader = doQuery();
                LOG.info("meta role: Observer, leaderInfo: {}", leader);
            } else {
                leader = doElect();
            }

            if (amILeader()) {
                onIamLeader();
            } else { // 我不是leader
                onIamNotLeader();
            }
        }
    }

    @Override
    public String myself() {
        return address;
    }


    @Override
    public synchronized void change2Follow() {
        this.startElector = true;
        this.isObserver = false;
    }


    @Override
    public synchronized void change2Observer() {
        this.isObserver = true;
    }


    protected abstract Leader doQuery();


    protected abstract Leader doElect();


    @Override
    public boolean amILeader() {
        return amILeader(leader.getLeader());
    }

    protected boolean amILeader(String leader) {
        return StringUtils.equals(myself(), leader) && leaderNotExpired();
    }

    private boolean leaderNotExpired() {
        long current = System.currentTimeMillis();
        return current < leader.getExpireTimestamp();
    }


    @Override
    public Leader getLeaderInfo() {

        if (leaderNotExpired()) {
            return leader;
        }
        return Leader.HAS_NO_LEADER;
    }


    private void onIamNotLeader() {
        for (LeaderAware leaderAware : leaderAwareList) {
            leaderAware.followNotify();
        }
    }


    private void onIamLeader() {
        for (LeaderAware leaderAware : leaderAwareList) {
            leaderAware.leaderNotify();
        }
    }

    private class LeaderElectorTrigger extends LoopExecuteTask {

        @Override
        public void _execute_() {
            if (startElector) {
                try {
                    elect();
                    LOG.info("after elect, leader={}", leader);
                } catch (Throwable e) {
                    LOG.error("failed to do elect", e);
                }
            }
        }

        @Override
        public void _wait_() {
            ConcurrentUtils.sleepUninterruptibly(1, TimeUnit.SECONDS);
        }
    }
}
